
package com.springboot.rmq.consumer.manager;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.springboot.rmq.comm.Constant;
import com.springboot.utils.SpringUtils;
import com.springboot.rmq.consumer.MsgConsumer;
import com.springboot.rmq.consumer.MsgHandle;
import com.springboot.rmq.consumer.bean.MsgConfigBean;
import com.springboot.rmq.consumer.meta.MsgMeta;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
import org.springframework.core.type.classreading.CachingMetadataReaderFactory;
import org.springframework.core.type.classreading.MetadataReader;
import org.springframework.core.type.classreading.MetadataReaderFactory;
import org.springframework.util.StringUtils;
import org.springframework.util.SystemPropertyUtils;

import java.io.IOException;
import java.lang.annotation.Annotation;
import java.util.*;



public class MsgConsumerManager {

    /**
     * 获取日志记录器
     */
    private Log log = LogFactory.getLog(MsgConsumerManager.class);

    /**
     * 自动注入ConsumerConfig
     */
    private Properties consumerProp;


    private List<MsgConsumer> msgConsumers = Lists.newArrayList();

    /**
     * 对注解内容进行处理
     *
     * @param msgMeta 注解信息
     * @param prop    默认配置文件
     * @return return value
     */

    private static MsgConfigBean dealMetaInfo(MsgMeta msgMeta, Properties prop) {

        MsgConfigBean msgConfigBean = new MsgConfigBean();
        //从配置文件中获取默认配置信息
        String nameAddr = prop.getProperty(Constant.KEY_CONSUMER_NAMESERVER);
        //String defaultTimeOutMints = prop.getProperty(Constant.DEFAULT_CONFIG_CONSUMER_TIMEOUTMINUTES);

        msgConfigBean.setNameAddr(nameAddr);
        msgConfigBean.setGroupName(msgMeta.group());
        msgConfigBean.setToppic(msgMeta.topic());
        msgConfigBean.setTags(msgMeta.tags());
        msgConfigBean.setKey(msgMeta.key());
        msgConfigBean.setInstanceName(msgMeta.instanceName());
        msgConfigBean.setSelfConsumer(msgMeta.selfConsumer());//是否本机消费

        //生成beankey，用来进行重复判断
        StringBuilder consumerInfoKey = new StringBuilder();
        consumerInfoKey.append(msgMeta.group()).append(Constant.PUNCTUATION_UNDERLINE)
                .append(msgMeta.topic()).append(Constant.PUNCTUATION_UNDERLINE).append(msgMeta.tags())
                .append(Constant.PUNCTUATION_UNDERLINE).append(msgMeta.instanceName());
        msgConfigBean.setBeanFlag(consumerInfoKey.toString());

        msgConfigBean.setOrder(msgMeta.isOrderly());

        //读取非必填的metadata元素，使用配置文件中的默认值
        int defaultThreadMin = Integer.parseInt(prop.getProperty(Constant.KEY_CONSUMER_THREADCOUNTMIN));

        int defaultThreadMax = Integer.parseInt(prop.getProperty(Constant.KEY_CONSUMER_THREADCOUNTMAX));

        int defaultBatchSize = Integer.parseInt(prop.getProperty(Constant.KEY_CONSUMER_BATCHSIZE));
        if (msgMeta.consumerThreadCountMin() <= 0 || msgMeta.consumerThreadCountMin() > defaultThreadMin) {
            msgConfigBean.setConsumerThreadCountMin(defaultThreadMin);
        } else {
            msgConfigBean.setConsumerThreadCountMin(msgMeta.consumerThreadCountMin());
        }
        if (msgMeta.consumerThreadCountMax() <= 0 || msgMeta.consumerThreadCountMax() > defaultThreadMax) {
            msgConfigBean.setConsumerThreadCountMax(defaultThreadMax);
        } else {
            msgConfigBean.setConsumerThreadCountMax(msgMeta.consumerThreadCountMin());
        }

        if (msgMeta.consumeMessageBatchMaxSize() <= 0 || msgMeta.consumeMessageBatchMaxSize() > defaultBatchSize) {
            msgConfigBean.setConsumeMessageBatchMaxSize(defaultBatchSize);
        } else {
            msgConfigBean.setConsumeMessageBatchMaxSize(msgMeta.consumeMessageBatchMaxSize());
        }
        if (StringUtils.isEmpty(msgMeta.instanceName())) {
            msgConfigBean.setInstanceName(prop.getProperty(Constant.KEY_CONSUMER_INSTANCE));
        }


        return msgConfigBean;
    }

    /**
     * 查找所有实现了指定注解的类
     *
     * @param scanPackages 待扫描路径
     * @param annotation   注解类
     * @return clazzSet 指定路径下，符合条件的类的集合
     */
    public Set<Class<? extends MsgHandle>> findAnnotationClass(String scanPackages, Class<? extends Annotation> annotation) {
        //获取指定包路径下的所有类名称
        Set<String> fullClazzSet = findPackageClass(scanPackages);
        log.info("load all class name from sanpackage over.");
        Set<Class<? extends MsgHandle>> classSet = new HashSet<Class<? extends MsgHandle>>();
        //根据类的完成路径，进行反射，并判断该类是否使用指定的annotation注解
        for (String clazz : fullClazzSet) {
            try {
                if (clazz.contains("cpm.mobcb.platform.BaseQueryModel")) {
                    String s = "1";
                }
                Class<?> clz = Class.forName(clazz);
                Annotation tempAnno = clz.getAnnotation(annotation);
                if (tempAnno != null) {
                    if (log.isDebugEnabled()) {
                        log.debug("add:[" + clazz + "] into reflect collection.");
                    }

                    classSet.add(clz.asSubclass(MsgHandle.class));
                }
            } catch (Exception e) {
                log.error("reflect class exception:", e);

            }
        }
        return classSet;
    }

    /**
     * 根据扫描包的,查询下面的所有类
     *
     * @param scanPackages 扫描的package路径
     * @return clazzSet 指定包下所有类的集合
     */
    public Set<String> findPackageClass(String scanPackages) {

        //根据每个服务中的message_producer.properties验证及排重包路径,避免父子路径多次扫描
        Set<String> packages = checkPackage(scanPackages);
        //使用spring的工具类来读取metadata信息
        ResourcePatternResolver resourcePatternResolver = new PathMatchingResourcePatternResolver();
        MetadataReaderFactory metadataReaderFactory = new CachingMetadataReaderFactory(resourcePatternResolver);
        Set<String> clazzSet = new HashSet<String>();
        //获取指定包路径下的所有的类名称的集合
        for (String basePackage : packages) {
            if (StringUtils.isEmpty(basePackage)) {
                continue;
            }
            String packageSearchPath = ResourcePatternResolver.CLASSPATH_ALL_URL_PREFIX
                    + org.springframework.util.ClassUtils.convertClassNameToResourcePath(
                    SystemPropertyUtils.resolvePlaceholders(basePackage))
                    + "/" + Constant.DEFAULT_RESOURCE_PATTERN;
            try {
                Resource[] resources = resourcePatternResolver.getResources(packageSearchPath);
                for (Resource resource : resources) {
                    //检查resource，这里的resource都是class
                    String clazz = loadClassName(metadataReaderFactory, resource);
                    if (null == clazz) {
                        continue;
                    }
                    clazzSet.add(clazz);
                }
            } catch (Exception e) {
                log.error("Get All classes in package :[" + basePackage + "] has some errors.", e);
            }

        }
        return clazzSet;
    }

    /**
     * 排重、检测package父子关系，避免多次扫描(包名可以指定多个，以逗号分割)
     *
     * @param scanPackages 带检查的包路径
     * @return 返回检查后有效的路径集合
     */
    private Set<String> checkPackage(String scanPackages) {

        Set<String> packages = new HashSet<String>();
        //排重路径,以逗号分割所有的路径
        Collections.addAll(packages, scanPackages.split(Constant.PUNCTUATION_COMMA));
        for (String innerAddr : packages.toArray(new String[packages.size()])) {
            //排除不符合要求的路径
            if (StringUtils.isEmpty(innerAddr) || innerAddr.equals(Constant.PUNCTUATION_PERIOD)
                    || innerAddr.startsWith(Constant.PUNCTUATION_PERIOD)) {
                continue;
            }
            //如果以句号结尾，则将句号截取掉
            if (innerAddr.endsWith(Constant.PUNCTUATION_PERIOD)) {
                innerAddr = innerAddr.substring(0, innerAddr.length() - 1);
            }
            Iterator<String> packageIte = packages.iterator();
            boolean needAdd = true;
            while (packageIte.hasNext()) {
                String pack = packageIte.next();
                if (innerAddr.startsWith(pack + Constant.PUNCTUATION_PERIOD)) {
                    //如果待加入的路径是已经加入的pack的子集，不加入
                    needAdd = false;
                } else if (pack.startsWith(innerAddr + Constant.PUNCTUATION_PERIOD)) {
                    //如果待加入的路径是已经加入的pack的父集，删除已加入的pack
                    packageIte.remove();
                }
            }
            if (needAdd) {
                packages.add(innerAddr);
            }
        }
        //返回对配置文件中包路径处理之后的路径
        return packages;
    }

    /**
     * 加载资源，根据resource获取className
     *
     * @param metadataReaderFactory spring中用来读取resource为class的工具
     * @param resource              这里的资源就是一个Class
     * @return fanhuizhi
     *
     * @throws IOException ioexception
     */
    private String loadClassName(MetadataReaderFactory metadataReaderFactory, Resource resource)
            throws IOException {
        try {
            if (resource.isReadable()) {
                MetadataReader metadataReader = metadataReaderFactory.getMetadataReader(resource);
                if (metadataReader != null) {
                    return metadataReader.getClassMetadata().getClassName();
                }
            }
        } catch (Exception e) {
            log.error("load class error.", e);
            return null;
        }
        return null;
    }

    /**
     * 扫描指定的包名，获取所有的符合条件的注解的类
     *
     * @return 返回值
     */
    public List<MsgConfigBean> genConsumerConfigList() {
        log.info("start to config consumer info.");
        //Properties prop = PropUtils.loadPropConfig(Constant.CONSUMER_CONFIG_NAME);
        Properties prop = consumerProp;
        String packages = prop.getProperty(Constant.KEY_CONSUMER_SCANPACKAGE);
        if (StringUtils.isEmpty(packages)) {
            log.error("scanpackage config is empty.");
            return null;
        }
        log.info("scan the package:[" + packages + "]");
        Map<String, MsgConfigBean> consumerConfigs = new HashMap<String, MsgConfigBean>();
        //查找所有使用了MsgMeta的类
        Set<Class<? extends MsgHandle>> classSet = findAnnotationClass(packages, MsgMeta.class);
        log.info("Reflect All Class Over.total class number:[" + classSet.size() + "]");
        String[] classNameArray=null;
        String shortClassName=null;
        String isNotAllowed=null;
        //遍历集合
        for (Class<? extends MsgHandle> tempClass : classSet) {
            //判断启动参数是否禁用了某个消费者，如果禁用了，不启动对应消费者
            //主要场景，批量发券的consumer会影响性能，单独放在一个ServiceGo上处理，其它ServiceGo处理正常业务

             classNameArray = tempClass.getName().split("\\.");
             shortClassName= classNameArray[classNameArray.length-1];//取类名的最后一段
             isNotAllowed = System.getProperty(shortClassName); //该Consumer是否被启动参数中禁用了
            if (!Strings.isNullOrEmpty(isNotAllowed)&& "false".equals(isNotAllowed)) {
                log.info("warning : class:[" + shortClassName + "] was not allowd to start! ");
                continue;
            }

        //判断是否实现了ConsumerMsgMeta注解信息
            MsgMeta meta = tempClass.getAnnotation(MsgMeta.class);
            if (meta == null) {
                log.info("class:[" + tempClass.getName() + "] meta info is wrong ");
                continue;
            }
            log.info("class:[" + tempClass.getName() + "] is invoked.");
            //通过注解的方式，注册监听类
            MsgConfigBean msgConfigBean = dealMetaInfo(meta, prop);
            msgConfigBean.setHandle(tempClass);
            try {
                // 有些代码可能没有使用Spring注解修饰Handler类
                msgConfigBean.setMsgHandle(SpringUtils.getBean(tempClass));
            } catch (Exception e) {
                log.warn("msgHandle not in SpringBeanFactory", e);
            }
            //读取符合条件的类中的注解字段配置信息
            if (consumerConfigs.containsKey(msgConfigBean.getBeanFlag())) {
                consumerConfigs.remove(msgConfigBean.getBeanFlag());
            }

            consumerConfigs.put(msgConfigBean.getBeanFlag(), msgConfigBean);
        }
        log.info("final consumer info size :[" + consumerConfigs.size() + "] and configure consumer over.");
        return new ArrayList<MsgConfigBean>(consumerConfigs.values());
    }

    public void initMsgHandler() {
        List<MsgConfigBean> configBeans = genConsumerConfigList();
        for (MsgConfigBean configBean : configBeans) {
            msgConsumers.add(new MsgConsumer(configBean));
        }
    }

    public Properties getConsumerProp() {
        return consumerProp;
    }

    public void setConsumerProp(Properties consumerProp) {
        this.consumerProp = consumerProp;
    }
}
